package cn.ucox.web.ms.polling;

import cn.ucox.web.ms.Bootstarpable;
import cn.ucox.web.ms.config.DataExchangeConfig.PollingConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.ref.WeakReference;
import java.util.List;

/**
 * 数据库扫描组件
 *
 * @author chenw
 * @create 2017-05-11 10:45
 * @email javacspring@gmail.com
 */
public class DataScanner implements Bootstarpable{

    private static final Logger logger = LoggerFactory.getLogger(DataScanner.class);

    private IDataLoader mDataLoader;
    private PollingConfig mPollingConfig;
    private BucketDataProcessor mDataProcessor;

    private static boolean isScannerRunning = false;
    private static long sDBPollInterval;
    private ScanningTask mScanningTask;

    public DataScanner(PollingConfig pollingConfig,
                       IDataLoader dataLoader,
                       BucketDataProcessor bucketDataProcessor) {
        this.mPollingConfig = pollingConfig;
        this.mDataLoader = dataLoader;
        this.mDataProcessor = bucketDataProcessor;
        mScanningTask = new ScanningTask(this);
        sDBPollInterval = mPollingConfig.getInterval();
    }

    /**
     * 开始启动扫描任务
     */
    @Override
    public void start() {
        if (null == mScanningTask) {
            mScanningTask = new ScanningTask(this);
        }
        if (isScannerRunning) {
            return;
        }
        mScanningTask.start();
        isScannerRunning = true;
    }

    @Override
    public void stop() {
        isScannerRunning = false;
        logger.info("数据加载组件停止");
    }

    private static class ScanningTask extends Thread {

        private WeakReference<DataScanner> weakReference;

        public ScanningTask(DataScanner scanner) {
            weakReference = new WeakReference<>(scanner);
        }

        private boolean preTimeHashData = true;

        @Override
        public void run() {
            do {
                DataScanner scanner = weakReference.get();
                if (null == scanner)
                    return;
                //从Redis中读取上次同步数据
                SyncAnchor anchor = SyncRedisUtil.loadSyncAnchorFromRedis();
                //检测FTP同步目录文件是否为0，否则提示存在文件滞留
                if (scanner.mPollingConfig.getMaxFtpDetention() < anchor.getMaxFtpDetention()) {
                    logger.debug("FTP滞留{}等待处理,延时{}ms检测任务完成情况", anchor.getMaxFtpDetention(), scanner.mPollingConfig.getInterval() / 2);
                    try {
                        Thread.sleep(scanner.mPollingConfig.getInterval() / 2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }


                //判断当前数据处理任 务是否已经完成，如果上一次事务数据未处理完成，则延时加载最新一次数据，延时时间为正常轮询时间一半
                if (anchor.isTransactionDone()) {
                    //重置同步标记,但不能重置同步数据结束时间，否则增量同步出错
                    anchor.reset();
                } else {
                    sDBPollInterval += scanner.mPollingConfig.getIntervalOffset();
                    if (sDBPollInterval > scanner.mPollingConfig.getMaxInterval())
                        sDBPollInterval = scanner.mPollingConfig.getMaxInterval();
                    logger.debug("数据正在处理中，：延时{}ms检测任务完成情况", scanner.mPollingConfig.getInterval() / 2);
                    try {
                        Thread.sleep(scanner.mPollingConfig.getInterval() / 2);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                //根据上次同步锚点记录，从数据库中加载最新数据
                DataBucket bucket = null;
                try {
                    bucket = scanner.mDataLoader.getData(anchor);
                } catch (Exception ex) {
                    logger.error("获取数据异常:{}", ex.getMessage(), ex);
                }
                //无数据则下次轮询时间延长offset
                if (null == bucket) {
                    sDBPollInterval += scanner.mPollingConfig.getIntervalOffset();
                    if (sDBPollInterval > scanner.mPollingConfig.getMaxInterval())
                        sDBPollInterval = scanner.mPollingConfig.getMaxInterval();
                    preTimeHashData = false;
                    logger.info("暂无同步数据，当前轮询间隔：{}ms", sDBPollInterval);
                } else {
                    List<SyncData> syncData = bucket.getSyncData();
                    //动态调频
                    if (preTimeHashData) { //上一次有数据，则降频
                        sDBPollInterval -= scanner.mPollingConfig.getIntervalOffset();
                        if (sDBPollInterval < scanner.mPollingConfig.getMinInterval())
                            sDBPollInterval = scanner.mPollingConfig.getMinInterval();
                    } else {//上一次没有数据，此次有数据，则调频至正常频率
                        sDBPollInterval = scanner.mPollingConfig.getInterval();
                    }
                    //设置同步锚点数据起始及结束时间
                    anchor.setDataStartTime(syncData.get(0).getTime());
                    anchor.setDataEndTime(syncData.get(syncData.size() - 1).getTime());
                    //设置同步任务起始结束时间
                    anchor.setSyncStartTime(bucket.getSyncStartTime());
                    anchor.setSyncEndTime(bucket.getSyncEndTime());

                    //数据处理
                    anchor.setTransProcessDone("N");//标记数据未处理完成
                    anchor.setTransactionId(anchor.getTransactionId() + 1);
                    SyncRedisUtil.updateSyncAnchor(anchor);
                    scanner.mDataProcessor.postBucketData(anchor, syncData);
                    preTimeHashData = true;//标记上一次有数据
                    logger.debug("数据投递至处理器，事务ID:{}, 当前轮询间隔:{}ms", anchor.getTransactionId(), sDBPollInterval);
                    //将此次同步锚点信息记录至Redis
                    logger.debug("刷新同步锚点成功");
                }
                try {
                    Thread.sleep(sDBPollInterval);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } while (isScannerRunning);
        }
    }
}
